package cn.doitedu.rtdw.dashboard;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job06_最近一小时的各类订单数 {

    public static void main(String[] args) {

        // 创建环境
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建业务库中的oms_order表的cdc连接器表
        tenv.executeSql(
                        " create table oms_order_mysql_cdc(                 \n "+
                        "   id           bigint,                            \n "+
                        " 	total_amount decimal(10,2),                     \n "+
                        " 	pay_amount   decimal(10,2),                     \n "+
                        " 	status       int,                               \n "+
                        " 	create_time  timestamp(3),                      \n "+  // 订单创建时间
                        " 	modify_time  timestamp(3),                      \n "+
                        " 	rt as modify_time,                              \n "+
                        " 	watermark for rt as rt - interval '0' second,   \n "+
                        "   primary key (id) not enforced                   \n "+
                        " )WITH (                                           \n "+
                        "     'connector' = 'mysql-cdc',                    \n "+
                        "     'hostname' = 'doitedu',                       \n "+
                        "     'port' = '3306',                              \n "+
                        "     'username' = 'root',                          \n "+
                        "     'password' = 'root',                          \n "+
                        "     'database-name' = 'realtimedw',               \n "+
                        "     'table-name' = 'oms_order'                    \n "+
                        " )                                                 \n "
        );

        // 创建mysql目标表的jdbc连接器表

        // 核心逻辑：累计窗口
//        tenv.executeSql(
//                        "                                                              "+
//                        " SELECT                                                       "+
//                        "     HOP_START(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR)     "+
//                        "     ,HOP_END(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR)      "+
//                        " 	  ,count(1) filter(where create_time > HOP_START(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR) AND status=0) as `待支付订单数`  "+
//                        " 	  ,count(1) filter(where create_time > HOP_START(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR) AND status=1) as `已支付订单数`  "+
//                        " 	  ,count(1) filter(where create_time > HOP_START(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR) AND status=2) as `待收货订单数`  "+
//                        " 	  ,count(1) filter(where create_time > HOP_START(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR) AND status=3) as `已完成订单数`  "+
//                        " 	  ,count(1) filter(where create_time > HOP_START(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR) AND status=4) as `已关闭订单数`  "+
//                        " 	  ,count(1) filter(where create_time > HOP_START(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR) AND status=5) as `已取消订单数`  "+
//                        " from oms_order_mysql_cdc                                     "+
//                        " group by                                                     "+
//                        "     HOP(rt,INTERVAL '1' SECOND,INTERVAL '1' HOUR)            "
//        ).print();

        tenv.executeSql(
                "                                                                       "+
                        " SELECT                                                       "+
                        "     HOP_START(rt,INTERVAL '1' MINUTE,INTERVAL '1' HOUR)      "+
                        "     ,HOP_END(rt,INTERVAL '1' MINUTE,INTERVAL '1' HOUR)       "+
                       " 	  ,count(1)     as cnt                                     "+
                       " 	  ,current_timestamp                                       "+ // 窗口触发时的系统时间（处理时间）
                        " from oms_order_mysql_cdc                                     "+
                        " group by                                                     "+
                        "     HOP(rt,INTERVAL '1' MINUTE,INTERVAL '1' HOUR)            "
        ).print();


    }

}
